package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.heima.article.service.ApArticleService;
import com.heima.article.service.HotArticleService;
import com.heima.behavior.BehaviorDto;
import com.heima.behavior.ReadBehaviorDto;
import com.heima.common.constants.ArticleConstants;
import com.heima.common.constants.HotArticleConstants;
import com.heima.common.constants.WemediaConstants;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;


@Component
public class KafkaCustomer {

    @Autowired
    private ApArticleService articleService;

    @Autowired
    private HotArticleService hotArticleService;

    /**
     * 监听文章上下架
     * @param message 监听发送的消息
     */
    @KafkaListener(topics = WemediaConstants.ARTICLE_DOWN_OR_UP)
    public void listenArticleDownOrUp(String message) {
        //参数校验
        if (StringUtils.isEmpty(message)) {
            return;
        }
        System.out.println("接收到消息" + message);
        //接收对象
        JSONObject article = JSON.parseObject(message);
        //修改app文章状态
        articleService.downOrUp(
                Long.valueOf(article.getLong("articleId") + ""),
                article.getShort("articleEnable"));
    }

    /**
     * 监听文章点赞消息
     * @param message 监听发送的消息
     */
    @KafkaListener(topics = ArticleConstants.ARTICLE_LIKE_BEHAVIOR)
    public void listenArticleLikeBehavior(String message) {
        //参数校验
        if (StringUtils.isEmpty(message)) {
            return;
        }
        System.out.println("接收到文章点赞消息" + message);
        //接收对象
        JSONObject jsonObject = JSONObject.parseObject(message);
        BehaviorDto behaviorDto = jsonObject.getObject("behaviorDto", BehaviorDto.class);
        Long userId = jsonObject.getLong("userId");
        //文章点赞
        articleService.likeBehavior(behaviorDto,userId);
    }

    /**
     * 监听文章阅读消息
     * @param message 监听发送的消息
     */
    @KafkaListener(topics = ArticleConstants.ARTICLE_READ_BEHAVIOR)
    public void listenArticleReadBehavior(String message) {
        //参数校验
        if (StringUtils.isEmpty(message)) {
            return;
        }
        System.out.println("接收到文章阅读消息" + message);
        //接收对象
        //接收对象
        JSONObject jsonObject = JSONObject.parseObject(message);
        ReadBehaviorDto readBehaviorDto = jsonObject.getObject("readBehaviorDto", ReadBehaviorDto.class);
        Long userId = jsonObject.getLong("userId");
        //文章点赞
        articleService.readBehavior(readBehaviorDto,userId);
    }

    /**
     * 监听文章不喜欢消息
     * @param message
     */
    @KafkaListener(topics = ArticleConstants.UN_LIKE_BEHAVIOR)
    public void listenArticleUnLikeBehavior(String message) {
        //参数校验
        if (StringUtils.isEmpty(message)) {
            return;
        }
        System.out.println("接收到文章不喜欢消息" + message);
        //接收对象
        JSONObject jsonObject = JSONObject.parseObject(message);
        BehaviorDto behaviorDto = jsonObject.getObject("behaviorDto", BehaviorDto.class);
        Long userId = jsonObject.getLong("userId");
        //文章不喜欢
        articleService.unLikeBehavior(behaviorDto,userId);
    }

    /**
     * 监听文章统计计算消息
     * @param message
     */
    @KafkaListener(topics = HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC)
    public void listenArticleCompute(String message) {
        //参数校验
        if (StringUtils.isEmpty(message)) {
            return;
        }
        System.out.println("接收到文章统计计算消息--处理开始" + message);
        hotArticleService.updateHotArticle(message);
        System.out.println("接收到文章统计计算消息--处理结束" + message);

    }
}
